package main.java.com.chankudo.thread;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadLocalRandom;
import java.util.function.Function;
import java.util.function.Supplier;

public class TestThread {
    /**
     * 测试join方法，等待线程执行结束。
     * @param args
     */
    public static void testJoin(String[] args) {
    //public static void main(String[] args) {
        SimpleThread threadOne = new SimpleThread("Thread-one");
        SimpleThread threadTwo = new SimpleThread("Thread-two");
        
        System.out.println("threadOne isActive:" + threadOne.isAlive());
        System.out.println("threadTwo isActive:" + threadTwo.isAlive());
        System.out.println("--------------------------------------------");

        threadOne.start();
        System.out.println("threadOne isActive:" + threadOne.isAlive());
        threadTwo.start();
        System.out.println("threadTwo isActive:" + threadTwo.isAlive());
        System.out.println("--------------------------------------------");

        try {
            threadOne.join();
            threadTwo.join();
        } catch (InterruptedException e) {
            // 线程被中断
            e.printStackTrace();
        }
        System.out.println("threadOne isActive:" + threadOne.isAlive());
        System.out.println("threadTwo isActive:" + threadTwo.isAlive());
        System.out.println("--------------------------------------------");
    }


    public static void testSynchronized(String[] args) throws InterruptedException {
        SimpleWaitQueue queue = new SimpleWaitQueue(10);

        Thread consumerThread = new Thread(new Runnable(){

            @Override
            public void run() {
                while(true){
                    queue.get();
                    //Thread.yield();
                }
            }
            
        }, "consumer"); //消费者线程

        Thread producerThread = new Thread(new Runnable(){

            @Override
            public void run() {
                while(true){
                    queue.put();
                    //Thread.yield();
                }
            }
            
        }, "producer"); //生产者线程

        producerThread.start();
        consumerThread.start();

        // try {
        //     Thread.sleep(100000);
        // } catch (InterruptedException e) {
        //     // TODO Auto-generated catch block
        //     e.printStackTrace();
        // }

        producerThread.join();
        consumerThread.join();

        System.out.println("exit");
    }

    public static void testLock(String[] args) throws InterruptedException {
        LockWaitQueue queue = new LockWaitQueue(10);

        int consumerSize = 10;
        int producerSize = 2;

        Runnable producerRunnable = ()->{
            try{
                while(!Thread.currentThread().isInterrupted() && true){
                    queue.put();
                }
            }catch(InterruptedException e) {
                // TODO Auto-generated catch block
            e.printStackTrace();
            }
        };
        Thread [] producerThreads = new Thread[producerSize];
        for(int i = 0; i < producerSize; i++){
            Thread producerThread = new Thread(producerRunnable, "producer-"+i);
            producerThreads[i]=producerThread;
            producerThread.start();
        }

        Runnable consumerRunnable = ()->{
            try{
               while(!Thread.currentThread().isInterrupted() && true){
                    queue.get();
                } 
            }catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                e.printStackTrace();
            }
        };
        Thread [] consumerThreads = new Thread[consumerSize];
        for(int i=0; i < consumerSize; i++){
            Thread consumerThread = new Thread(consumerRunnable, "consumer-"+i);
            consumerThreads[i] = consumerThread;
            consumerThread.start();
        }
        
        // 守护进程，当其他进程都结束时，守护进程也将自动结束。
        //consumerThread.setDaemon(true);
        //producerThread.setDaemon(true);

        for(Thread producer : producerThreads){
            producer.join();
        }

        for(Thread consumer :consumerThreads){
            consumer.join();
        }

        System.out.println("exit");

    }

    /**
     * 阻塞队列的例子
     * @param args
     */
    public static void testQueue(String[] args) {
        // 每个子线程都独立的使用局部变量，以保证此变量的线程安全。
        // 注意：此时get()方法返回的变量时各子线程独有的，各子线程不再共享此局部变量。
        //final ThreadLocal<Random> tRandom = ThreadLocal.withInitial(()-> new Random());
        // 随机数有更方便的方法：ThreadLocalRandom.current() 直接返回特定于当前线程的Radmon的实例

        //队列大小
        int capacity = 100;

        //有指定容量和公平性设置的阻塞队列 
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(capacity);
        // 常用的阻塞队列
        // ArrayBlockingQueue 实现为循环数组，初始化时需要指定容量。
        // LinkedBlockingQueue 队列，实现为链表，初始化时容量参数不是必须的参数。
        // LinkedBlockingDeque 双向队列，实现为链表，初始化时容量参数不是必须的参数。

        // 生产者队列数量
        int producerCount = 2;
        Thread[] producerThreads = new Thread[producerCount];
        Runnable producerRunnable =  new Runnable(){

            @Override
            public void run() {
                while(!Thread.currentThread().isInterrupted() && true){
                    //int temp = tRandom.get().nextInt();
                    int temp = ThreadLocalRandom.current().nextInt();
                    System.out.println(Thread.currentThread().getName()+" put " + temp); 
                    try {
                        queue.put(temp);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }

        };
        for (int i = 0; i < producerCount; i++){
            Thread thread = new Thread(producerRunnable, "producer-"+i);
            producerThreads[i] = thread;
            thread.start();
        }

        // 消费者队列数量
        int consumerCount = 5;
        Thread[] consumerThreads = new Thread[consumerCount];
        Runnable consumeRunnable = new Runnable(){

            @Override
            public void run() {
                while(!Thread.currentThread().isInterrupted() && true){
                    try {
                        int temp = queue.take();
                        System.out.println(Thread.currentThread().getName() + " take " + temp);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                
            }
        };
        for(int j = 0; j < consumerCount; j++){
            Thread thread = new Thread(consumeRunnable, "consumer-" + j);
            consumerThreads[j] = thread;
            thread.start();
        }

        System.out.println("exit");
    }

    /**
     * 学习使用执行器来执行 callable
     * @param args
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public static void testExecutor(String[] args) throws InterruptedException, ExecutionException {
        // callable 任务
        SimpleCall simpleCall = new SimpleCall();

        // 从执行器中获取线程池
        // 注意常用的线程池的区别:newSingleThreadExecutor newCachedThreadPool newFixedThreadPool newSingleThreadScheduledExecutor
        ExecutorService executorService = Executors.newSingleThreadExecutor();

        // 向线程池提交任务，返回 future 以获取执行结果
        Future futureTask = executorService.submit(simpleCall);

        // 可以在这里做其他的事情
        System.out.println("main thread: do something ...");

        // 需要的时候再获取
        Integer result = (Integer) futureTask.get();

        System.out.println("main thread get:" + result);

        // 执行此方法后，线程池不再接受新的任务了。
        executorService.shutdown();

    }

    /**
     * 学习使用线程池实现多线程执行
     * @param args
     */
    public static void testExecutorService(String[] args) {
        int callableNums = 5;
        List<Callable<Integer>> callabList = new ArrayList<Callable<Integer>>();

        // 初始化一组callalbe任务
        for(int i =0; i<callableNums; i++){
            // callable 任务
            SimpleCall simpleCall = new SimpleCall();
            callabList.add(simpleCall);
        }
        
        // 获取一个线程池
        ExecutorService executorService = Executors.newCachedThreadPool();

        // 法一 执行器顺序获取所有的callable结果
        // 顺序并且逐一获取执行结果
        // List<Future>exeResults = new ArrayList<>();
        // try {
        //     exeResults.addAll( executorService.invokeAll(callabList) ) ;
        // } catch (InterruptedException e) {
        //     // TODO Auto-generated catch block
        //     e.printStackTrace();
        // }
        
        // for(Future exeResult :exeResults){
        //         try {
        //             int result = (int) exeResult.get();
        //             System.out.println("main thread get reslut:"+result);
        //         } catch (InterruptedException | ExecutionException e) {
        //             // TODO Auto-generated catch block
        //             e.printStackTrace();
        //         }
        // }

        //法二：跑一组任务，但只要返回一个执行结果就结束所有任务。
        try {
            Integer result =  executorService.invokeAny(callabList);
            System.out.println("main thread get reslut:"+result);
        } catch (InterruptedException | ExecutionException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

        // // 法三：构造一个执行器服务，收集返回的结果。优先取最先拿到的结果
        // ExecutorCompletionService completionService = new ExecutorCompletionService<>(executorService);
        // // 逐一提交任务
        // for(Callable<Integer>  callable: callabList){
        //     completionService.submit(callable);
        // }
        // // 先结束的任务的执行结果放到一个阻塞队列中
        // // 优先得到最快执行完的任务
        // for(int i =0; i<callableNums; i++){
        //     try {
        //         int result = (int) completionService.take().get();
        //         System.out.println("main thread get reslut:"+result);
        //     } catch (InterruptedException | ExecutionException e) {
        //         // TODO Auto-generated catch block
        //         e.printStackTrace();
        //     }
        // }

        executorService.shutdown();
        System.out.println("exit.");
    }

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        /*第一部分：和之前一样 启动多线程任务，并等待任务执行结果 */
        // 与callabel类似
        // 异步执行，返回有结果的方法。

        Supplier demoSupplier = new Supplier<String>(){

            @Override
            public String get() {
                int temp = ThreadLocalRandom.current().nextInt(3000);
                try {
                    Thread.currentThread().sleep(temp);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                return Thread.currentThread().getName()+"-has sleeped "+temp;
            }
            
        };
        CompletableFuture<String> future =  CompletableFuture.supplyAsync(demoSupplier);
        // System.out.println("main thread get :"+future.get());

        // 也可执行runnable
        // 执行不需要返回结果的任务

        // Runnable runnable = new Runnable(){

        //     @Override
        //     public void run() {
        //         int temp = ThreadLocalRandom.current().nextInt(3000);
        //         try {
        //             Thread.currentThread().sleep(temp);
        //         } catch (InterruptedException e) {
        //             e.printStackTrace();
        //         }
        //         System.out.println(Thread.currentThread().getName()+"-has sleeped "+temp);
        //     }
            
        // };

        // CompletableFuture future = CompletableFuture.runAsync(runnable);
        // // 等待子任务完成
        // System.out.println("main thread get :"+future.get());

        /*通过回调的方式实现异步*/

        // 接 supplyAsync 执行之后 继续执行的操作；
        // thenApply方法会让指定同一个线程继续执行后面的操作。
        CompletableFuture<String> thenApplyFuture = 
        future.thenApply((String result)->{
            String newResult = Thread.currentThread().getName() +": get result ->" +result;
            return newResult;
        });
        //System.out.println("main thread get :"+thenApplyFuture.get());

        // thenApplyAsync 方法会让任务放到线程池中，继续执行后面的操作的不一定时同一个线程。
        CompletableFuture<String> thenApplyAsyncFuture = 
        thenApplyFuture.thenApplyAsync((String result)->{
            String newResult = Thread.currentThread().getName() +": get result ->" +result;
            // if(ThreadLocalRandom.current().nextInt() %2 ==1){
            //     throw new RuntimeException("test");
            // }
            return newResult;
        });
        //System.out.println("main thread get :"+thenApplyAsyncFuture.get());

        // thenAccept 该方法的执行接上一步执行的结果，但是该方法执行完不返回任何结果
        // thenAcceptAsync 方法与 thenAccept 的区别 与 thenApply/thenApplyAsync的区别类似
        // CompletableFuture thenAcceptAsyncFuture =  
        // thenApplyAsyncFuture.thenAcceptAsync((String result)->{
        //     System.out.println("thenAcceptAsync method thread name = "+Thread.currentThread().getName()+ " get params =" + result);
        //     throw new RuntimeException("test");
        // });
        //System.out.println("main thread get :"+thenAcceptAsyncFuture.get());

        // thenRun方法 上一步执行完后执行这一步 区别在于 上一步和这一步执行的方法不带参数，并且也不返回任何结果执行
        // 具体例子不展示了

        // 如果上一步执行抛出异常时，抛出的异常会作为参数，进入到这个方法中。
        // CompletableFuture<String> exceptionallyFuture =
        // thenAcceptAsyncFuture.exceptionally((e)->{
        //     ((Throwable) e).printStackTrace();
        //     return "exceptionally end...";
        // });
        // System.out.println("main thread get :"+exceptionallyFuture.get());
        
        // CompletableFuture 当某个任务执行完成后的回调函数
        // 如果正常执行的话 取第一个参数
        // 如果异常执行的话 取第二个参数 获取异常信息
        // 注意：若上一步正常结束，get方法返回上一步的执行结果。如果上一步异常结束，get方法则会抛出异常
        // CompletableFuture whenCompleteFuture = thenApplyAsyncFuture.whenComplete((result,exp)->{
        //     if(exp ==null){ //如果程序正常执行，就获取上一步正常执行的结果
        //         System.out.println("normally invoke whenComplete");
        //         //System.out.println(result.toString());
        //     }else{//上一步执行异常结束，获取异常信息。
        //         System.out.println("abnormally invoke whenComplete");
        //         //((Throwable) exp).printStackTrace();
        //     }
        // });
        // System.out.println("main thread get :"+whenCompleteFuture.get());
        
        // handle方法与CompletableFuture方法大致类似
        // 区别在于handle返回的是回调方法的执行结果或回调方法执行期间抛出的异常，与上一步的执行结果无关了。
        // CompletableFuture<String> handleFuture = thenApplyAsyncFuture.handle((result, exp)->{
        //     if(exp ==null){ //如果程序正常执行，就获取上一步正常执行的结果
        //         System.out.println("normally invoke whenComplete");
        //         System.out.println(result.toString());
        //         return "success";
        //     }else{//上一步执行异常结束，获取异常信息。
        //         System.out.println("abnormally invoke whenComplete");
        //         ((Throwable) exp).printStackTrace();
        //         return "fail";
        //     }
        // });
        // System.out.println("main thread get :"+handleFuture.get());


        //todo 组合处理
        //有待深入研究

         /*
         1.thenCombine / thenAcceptBoth / runAfterBoth
         共同点：这三个方法都是将两个CompletableFuture组合起来，只有这两个都正常执行完了才会执行某个任务
         区别：
         thenCombine 会将 两个任务的执行结果 作为方法入参传递到指定方法中，且该方法 有返回值；
         thenAcceptBoth 同样将 两个任务的执行结果 作为方法入参，但是 无返回值；
         runAfterBoth 没有入参，也 没有返回值。
         注意：两个任务中只要有一个执行异常，则将该异常信息作为指定任务的执行结果。
         */
        // CompletableFuture<String> thenCombineFuture = future1.thenCombine(future2,
        // (a,b)->{
        //     return "hello";
        // } 
        // );

        /*
        2.applyToEither / acceptEither / runAfterEither
        共同点：这三个方法都是将两个CompletableFuture组合起来，只要其中一个执行完了就会执行某个任务
        区别：
        applyToEither 会将 已经执行完成的任务的执行结果 作为方法入参，并 有返回值；
        acceptEither 同样将 已经执行完成的任务的执行结果作为方法入参，但是 没有返回值；
        runAfterEither 没有方法入参，也没有返回值；
         注意：两个任务中只要有一个执行异常，则将该异常信息作为指定任务的执行结果。
        */
        
       
        // thenCompose方法会在某个任务执行完成后，将该 任务的执行结果 作为方法入参然后执行指定的方法，该方法会返回一个 新的CompletableFuture实例。
        // 调用get方法返回的是这个新的CompletableFuture实例的返回结果。
        CompletableFuture<String> thenComposeFuture = thenApplyAsyncFuture.thenCompose((params)->{
            System.out.println("in thenCompose do something ->" + params);
            if(ThreadLocalRandom.current().nextInt() %2 ==1){
                return CompletableFuture.supplyAsync(new Supplier<String>(){
                    @Override
                    public String get() {
                        System.out.println("abnormally invoke thenCompose");
                        return null;
                    }
                    
                });
            }
            else{
                return CompletableFuture.supplyAsync(new Supplier<String>(){
                    @Override
                    public String get() {
                        System.out.println("normally invoke thenCompose");
                        return "thenCompose success";
                    }
                    
                });
            }
        });
        System.out.println("main thread get :"+thenComposeFuture.get());

        // allOf / anyOf
        // allOf 返回的CompletableFuture是多个任务都执行完成后才会执行，
        // 只要有一个任务执行异常，则返回的CompletableFuture执行get方法时会抛出异常
        // 如果都是正常执行，则get返回null
        // anyOf 只要有一个任务执行完成即可
        // CompletableFuture allOfFuture= CompletableFuture.allOf(future1,future2,future3);
        // // 
        // CompletableFuture allOfallOfeFutureFuture = allOfFuture.whenComplete((a,b)->{
        //     if(b==null){//所有任务正常执行结束

        //     }else{//存在任务执行异常结束

        //     }
        // });
        // System.out.println("main thread get :"+allOfallOfeFutureFuture.get());



    }



}
